package sos.entrega2013.mr1.noviembre.bo.facturacion;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.StringTokenizer;

import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;

import sos.entrega2013.mr1.noviembre.beans.CDR;

public class FacturacionMap extends MapReduceBase 
	implements Mapper<LongWritable, Text, Text, CDR>{
	private SimpleDateFormat formatoDeFecha = new SimpleDateFormat("yyyyMMddHHmmss");
	private Text 			 tLlamante 		= new Text();
	private CDR				 cdrDatos	 	= new CDR();
	private Date		 	 dFecha		 	= null;
	private String 			 sNTelefono	 	= null;
	@Override
	public void map(LongWritable key, Text value,OutputCollector<Text, CDR> output, Reporter reporter)
	throws IOException {
		
		String sLine 				= value.toString();
		StringTokenizer tokenizer 	= new StringTokenizer(sLine,",");
		//------------------------------------------------------------
		// Leemos el telefono del llamante
		this.sNTelefono				= tokenizer.nextToken().trim();
		this.tLlamante.set(this.sNTelefono);
		// asignamos a el CDR el numero del llamante
		this.cdrDatos.setLlamante(this.sNTelefono);
		//------------------------------------------------------------
		// Leemos el telefono del llamado
		this.cdrDatos.setLlamado(tokenizer.nextToken().trim());
		//------------------------------------------------------------
		// Leemos la fecha
		try {
			this.dFecha=formatoDeFecha.parse(tokenizer.nextToken().trim());
		} catch (ParseException e) {e.printStackTrace();}
		// asignamos la fecha
		this.cdrDatos.setFecha(dFecha);
		//-------------------------------------------------------------
		this.cdrDatos.setDuracion(Integer.valueOf(tokenizer.nextToken().trim()));
		this.cdrDatos.setCoordXllamante(Integer.valueOf(tokenizer.nextToken().trim()));
		this.cdrDatos.setCoordYllamante(Integer.valueOf(tokenizer.nextToken().trim()));
		this.cdrDatos.setCoordXllamado(Integer.valueOf(tokenizer.nextToken().trim()));
		this.cdrDatos.setCoordYllamado(Integer.valueOf(tokenizer.nextToken().trim()));
		//--------------------------------------------------------------
		output.collect(this.tLlamante, this.cdrDatos);		
	}

}
